package com.shujia.core

import com.mysql.jdbc.Driver
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, ResultSet, Statement}

object Code06TextJoinMysql {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("Mysql2Text"))

    sc
      .textFile("spark_code/data/score_join.txt",3)
      .mapPartitions {
        case line => {
          // 1、连接Mysql
          DriverManager.registerDriver(new Driver())
          val connection: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/stu?useSSL=false", "root", "123456")
          val statement: Statement = connection.createStatement()

          val iterator: Iterator[String] = line.map {
            case oneLine => {
              val splitRes: Array[String] = oneLine.split(",")
              val id: String = splitRes(0)
              val courseID: String = splitRes(1)
              val score: String = splitRes(2)
              val set: ResultSet = statement.executeQuery(
                s"""
                   |SELECT
                   |id
                   |,name
                   |,age
                   |,gender
                   |,substr(clazz,1,4)
                   |FROM students
                   |WHERE id= $id
                   |""".stripMargin)

              //              val tuple = new Tuple7[String, String, String, String, String,String, String](_, _, _, _, _,_,_)

              var res = ""
              while (set.next()) {
                res = (set.getString(1)
                  , set.getString(2)
                  , set.getString(3)
                  , set.getString(4)
                  , set.getString(5)
                  , courseID
                  , score
                ).toString()
              }
              res
            }
          }
          iterator
        }
      }
      .foreach(println)
  }

}
